package com.wubai.bigdata.flink.sink.util;

import org.apache.http.ParseException;
import org.apache.http.util.EntityUtils;
import org.elasticsearch.client.*;
import org.apache.http.HttpEntity;
import org.apache.http.HttpStatus;
import org.apache.http.entity.ContentType;
import org.apache.http.nio.entity.NStringEntity;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Collections;
import java.util.Map;
/**
 * @since 8.2.0
 * @author hecq
 */
public class CreateIndex extends Thread {
    private static final Logger LOG = LoggerFactory.getLogger(CreateIndex.class);

    public static void index(RestClient restClient, String mappingPath, String indexNamePrefix) throws IOException {
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                while (true) {
                    long start = System.currentTimeMillis();
                    Calendar calendar = Calendar.getInstance();
                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy_MM");
                    String indexName = indexNamePrefix;
                    Map<String, String> params = Collections.singletonMap("pretty", "true");


                    try {

                        System.out.println("mappingPath: " + mappingPath);
                        String path = CreateIndex.class.getClassLoader().getResource(mappingPath).getPath();
                        BufferedReader bufferedReader = new BufferedReader(new FileReader(new File(path)));
                        StringBuffer stringBuffer = new StringBuffer();
                        String line = null;
                        while ((line = bufferedReader.readLine()) != null) {
                            stringBuffer.append(line + "\n\r");
                        }
                        HttpEntity entity = new NStringEntity(stringBuffer.toString(), ContentType.APPLICATION_JSON);

                        // 创建Request对象
                        Request request = new Request("PUT", "/" + indexName);
                        request.addParameter("pretty", "true");
                        request.setEntity(entity);
                        Response rsp = restClient.performRequest(request);
                        if (HttpStatus.SC_OK == rsp.getStatusLine().getStatusCode()) {
                            LOG.info("CreateIndexWithShardNum successful.");
                        } else {
                            LOG.error("CreateIndexWithShardNum failed.");
                        }
                        LOG.info("CreateIndexWithShardNum response entity is : {}.", EntityUtils.toString(rsp.getEntity()));
                    } catch (IOException | ParseException e) {
                        LOG.error("CreateIndexWithShardNum failed, exception occurred.", e);
                    }

                    calendar.add(Calendar.MONTH, 1);
                    calendar.set(Calendar.DAY_OF_MONTH, 1);
                    calendar.set(Calendar.HOUR_OF_DAY, 0);
                    calendar.set(Calendar.MINUTE, 0);
                    calendar.set(Calendar.SECOND, 0);
                    long sleep = calendar.getTime().getTime() - start;
                    try {
                        Thread.sleep(sleep);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        Thread thread = new Thread(runnable);
        thread.start();
    }
}